package dm.ass2.item;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Random;
import java.util.Set;

import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import dm.ass1.io.ProfileWritable;
import dm.ass1.io.RatingWritable;
import dm.ass2.io.CodebookItemsWritable;
import dm.ass2.io.CodebookUsersWritable;
import dm.ass2.item.io.ItemMapOutputValueWritable;
import dm.ass2.user.io.UserMapOutputValueWritable;

public class ItemDriver extends Configured implements Tool {
	private static final Log LOG = LogFactory.getLog(ItemDriver.class);
	
	public static final String NUM_OF_USER_CLUSTERS = "num.of.user.clusters";
	public static final String NUM_OF_ITEM_CLUSTERS = "num.of.item.clusters";

	Set<Integer> itemsIds;
	Set<Integer> usersIds;
	
	//int numberOfUsers = -1;
	//int numberOfItems = -1;
	
	int numberOfItemsClusters = 2;
	int numberOfUsersClusters = 2;
	
	public void setNumberOfClusters(int numberOfItemsClusters, int numberOfUsersClusters) {
		this.numberOfItemsClusters = numberOfItemsClusters;
		this.numberOfUsersClusters = numberOfUsersClusters;
	}
	
	private boolean runProfilingJob(String inputPath) throws Exception {
		Job job = new Job(new Configuration());
		Configuration conf = job.getConfiguration();
		conf.set(NUM_OF_ITEM_CLUSTERS, String.valueOf(this.numberOfItemsClusters));
		conf.set(NUM_OF_USER_CLUSTERS, String.valueOf(this.numberOfUsersClusters));
		
		FileSystem fs = FileSystem.getLocal(conf);
	    Path tempOutputPath = new Path("output/itemTemp/");
	    if (fs.exists(tempOutputPath)) {
	    	fs.delete(tempOutputPath,true);			
		}
	    		
		job.setJarByClass(ItemDriver.class);
		FileInputFormat.addInputPath(job, new Path(inputPath));
		job.setOutputFormatClass(SequenceFileOutputFormat.class);
		SequenceFileOutputFormat.setOutputPath(job, tempOutputPath);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(CodebookItemsWritable.class);
		job.setMapOutputKeyClass(IntWritable.class);
		job.setMapOutputValueClass(ItemMapOutputValueWritable.class);
		job.setMapperClass(ItemMap.class);
		job.setCombinerClass(ItemCombiner.class);
		job.setReducerClass(ItemReducer.class);
		
		return job.waitForCompletion(true);		
	}
	
	@Override
	public int run(String[] args) throws Exception {
		runProfilingJob(args[0]);
		
		return 0;

	}

	public static void main(String[] args) throws Exception {
		int ret = ToolRunner.run(new ItemDriver(), args);
		System.exit(ret);
	}
}
